/**
 * <p>文件名:		MessageServiceJMSImpl.java</p>
 * <p>版权:		CopyrightTag</p>
 * <p>公司:		千方集团CTFO</p>
 * @author		周华彬(zhouhuabin@ctfo.com, zhou_hua_bin@163.com)
 */

package com.caits.lbs.framework.services.sms;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.log4j.Logger;
import org.springframework.util.Assert;

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPubSub;
import redis.clients.util.SafeEncoder;

import com.caits.lbs.framework.log.CommonLogFactory;
import com.caits.lbs.framework.utils.StringUtils;

/**
 * <p>
 * MessageServiceJMSImpl
 * </p>
 * <p>
 * 用途：短消息发送的reids实现，利用redis的订阅机制
 * </p>
 * 
 * @author 周华彬(zhouhuabin@ctfo.com, zhou_hua_bin@163.com)
 * @version 0.0.1 2012-10-17
 *          <table style="border:1px solid gray;">
 *          <tr>
 *          <th width="100px">版本号</th>
 *          <th width="100px">动作</th>
 *          <th width="100px">修改人</th>
 *          <th width="100px">修改时间</th>
 *          </tr>
 *          <!-- 以 Table 方式书写修改历史 -->
 *          <tr>
 *          <td>0.0.0</td>
 *          <td>创建类</td>
 *          <td>zhb</td>
 *          <td>2012-10-17 下午5:31:11</td>
 *          </tr>
 *          <tr>
 *          <td>0.0.1</td>
 *          <td>创建类</td>
 *          <td>zhb</td>
 *          <td>2012-10-17 下午5:31:11</td>
 *          <td>0.0.2</td>
 *          <td>修改类</td>
 *          <td>xxx</td>
 *          <td>x年x月x日</td>
 *          </tr>
 *          </table>
 */
public class MessageServiceRedisImpl extends Thread implements IMessageService {

	/** 日志记录器 */
	private Logger log = CommonLogFactory.getLog();
	 
	/** 高速缓存服务地址 */
	private String serverAddress = "";
	
	/** 服务器验证密码 */
	private String password;
	/** 网络超时时间阈值ms */
	private int soTimeout=3*1000;

	/** 订阅的样式，例如new* */
	private String pattern;
	/** 消息消费者列表 */
	private final List<IConsumeMessage> consumeList = Collections.synchronizedList(new ArrayList<IConsumeMessage>());

	/** 服务器集群连接对象，便于退订 */
	private JedisCluster jmsReceiver = null;
	
	/** 保存消息订阅处理对象，便于退订 */
	private JedisPubSub subscriber = null;
	
	/** 接收线程是否运行 */
	private boolean bRun = false;
	
	/**
	 * 初始化方法, 在applicationContext-redis.xml 中 
	 * <bean id="redisCacheServer" class="com.caits.lbs.framework.services.memcache.RedisCacheManager"
	 * init-method="init"> 
	 * <property name="config" ref="jedisPoolConfig"></property>
	 * <property name="serverAddress" value="192.168.2.186:6379,192.168.4.115:6379"></property> 
	 * </bean>
	 * */
	public void init() {
		if(jmsReceiver!=null){
			throw new IllegalStateException("实例连接池pool已经初始化，放弃本次执行.");
		}
		/** 设置与缓存服务器的连接池 ,启用主备模式*/
		try {
			//
			Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();  
			String[] server_arr=StringUtils.splitToArray(serverAddress, ",", true);
			Assert.isTrue( server_arr.length>0,"方法init的serverAddress参数错误");
			for(String server:server_arr){
				String[]  hostPort=StringUtils.splitToArray(server, ":", true);
				if(hostPort.length>1){
					jedisClusterNodes.add(new HostAndPort(hostPort[0], Integer.valueOf(hostPort[1])));  
				}
			}
			jmsReceiver =new JedisCluster(jedisClusterNodes, soTimeout,5,new GenericObjectPoolConfig(),getPassword());
			log.info("MessageServiceRedisImpl initialization successful!"+serverAddress);
		} catch (Exception e) {
			log.error("初始化RedisCacheManager连接失败:" + e.getLocalizedMessage(), e);
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.caits.lbs.framework.services.sms.IMessageService#sendMessage(com.
	 * caits.lbs.framework.services.sms.IMessage)
	 */
	@Override
	public void sendMessage(final IMessage msg) {
		try {
			Assert.notNull(jmsReceiver, "消息发送对象pool不能为空.");
			Assert.notNull(msg, "消息内容msg不能为空.");
//			Jedis jedis = new Jedis();
//			jedis.publish(msg.getHead(), msg.getBody())
			jmsReceiver.publish(msg.getHead(), msg.getBody());
			log.info("消息发送成功,head:" + msg.getHead() + ",body:" + msg.getBody());
		} catch (Exception e) {
			log.error("消息发送失败,msg=" + msg.getBody(), e);
		}finally{
			
		}
	}

	/**
	 * 开始接收消息 
	 */
	public void startListen() {
		subscribePattern(getPattern());
	}

	/**
	 * 停止接收消息 
	 */
	public void stopListen() {
		bRun=false;
		if(subscriber!=null){
			subscriber.unsubscribe(getPattern());
			log.info("退订消息成功,channel=" + getPattern());
		}
	}
	/*
	 * (non-Javadoc)
	 * 
	 * @see java.lang.Thread#run()
	 */
	@Override
	public void run() {
		while (bRun) {
			synchronized (this) {
				try {
					wait();
				} catch (InterruptedException e) {
					log.error("thread interupted,msg="+e.getLocalizedMessage(), e);
				}
			}
		}
		log.info("接收消息线程结束.");
	}

	/**
	 * 订阅某个样式 
	 */
	protected void subscribePattern(String pat) {
		try {
			if(subscriber==null)
			subscriber=new JedisPubSub() {
				@Override
				public void onUnsubscribe(String channel, int subscribedChannels) {
					String chan =channel;
					log.info("取消订阅channel=" + chan+",subscribedChannels="+subscribedChannels);
						
				}
				
				@Override
				public void onSubscribe(String channel, int subscribedChannels) {
					String chan =channel;
					log.info("订阅消息成功channel=" + chan+",subscribedChannels="+subscribedChannels);
					
				}
				
				@Override
				public void onPUnsubscribe(String pattern, int subscribedChannels) {
					String pat =pattern;
					log.info("取消订阅channel=" + pat+",subscribedChannels="+subscribedChannels);
					
				}
				
				@Override
				public void onPSubscribe(String pattern, int subscribedChannels) {
					String pat =pattern;
					log.info("订阅channel=" + pat+",subscribedChannels="+subscribedChannels);
					
				}
				
				@Override
				public void onPMessage(String pattern, String channel, String message) {
					String msg =message;
					if (msg != null) {
						log.info("收到消息:" + msg+",from channel "+SafeEncoder.encode(channel)+",pattern="+SafeEncoder.encode(pattern));
						dispatchMessage(msg);
					}else{
						log.debug("没有收到消息,msg=null");
					}
				}
				
				@Override
				public void onMessage(String channel, String message) {
					String msg =message;
					String chan = channel;
					if (msg != null) {
						log.info("收到消息,channel=" +chan+",msg="+msg);
						MessageSMS sms= new MessageSMS();
						sms.setHead(chan);
						sms.setBody(msg);
						dispatchMessage(sms);
					}else{
						log.debug("没有收到消息,msg=null");
					}
				}
			};
			jmsReceiver.subscribe(subscriber,pat);
			
		} catch (Exception e) {
			log.error("接收消息失败,msg="+e.getLocalizedMessage()+",请重新调用subscribe订阅", e);
			
		}finally{
			if(jmsReceiver!=null)
			jmsReceiver=null;
		}
	}

	/* (non-Javadoc)
	 * @see com.caits.lbs.framework.services.sms.IMessageService#registerConsumer(com.caits.lbs.framework.services.sms.IConsumeMessage)
	 */
	@Override
	public void registerConsumer(IConsumeMessage consumer) {
		if(!consumeList.contains(consumer))
			consumeList.add(consumer);
	}

	/**
	 * 给消费者列表分配消息 
	 * @param message
	 */
	public void dispatchMessage(Object message) {
		for(IConsumeMessage consumer:consumeList){
			try{
				if(consumer!=null)consumer.onMessage(message);
			}catch(Exception e){
				log.error("消费者处理消息异常,consume="+consumer.hashCode(), e);
			}
		}
	}

	/**
	 * 获取变量<code>serverAddress</code>的值
	 * @return 返回的数据类型是<code>String</code>
	 */
	public String getServerAddress() {
		return serverAddress;
	}

	/**
	 * 设置变量<code> serverAddress</code> 的值
	 * @param serverAddress  <code>serverAddress</code> 参数类型是<code>String</code>
	 */
	public void setServerAddress(String serverAddress) {
		this.serverAddress = serverAddress;
	}

	/**
	 * 获取变量<code>password</code>的值
	 * @return 返回的数据类型是<code>String</code>
	 */
	public String getPassword() {
		return password;
	}

	/**
	 * 设置变量<code> password</code> 的值
	 * @param password  <code>password</code> 参数类型是<code>String</code>
	 */
	public void setPassword(String password) {
		this.password = password;
	}

	/**
	 * 获取变量<code>soTimeout</code>的值
	 * @return 返回的数据类型是<code>int</code>
	 */
	public int getSoTimeout() {
		return soTimeout;
	}

	/**
	 * 设置变量<code> soTimeout</code> 的值
	 * @param soTimeout  <code>soTimeout</code> 参数类型是<code>int</code>
	 */
	public void setSoTimeout(int soTimeout) {
		this.soTimeout = soTimeout;
	}

	/**
	 * 获取变量<code>pattern</code>的值
	 * @return 返回的数据类型是<code>String</code>
	 */
	public String getPattern() {
		return pattern;
	}

	/**
	 * 设置变量<code> pattern</code> 的值
	 * @param pattern  <code>pattern</code> 参数类型是<code>String</code>
	 */
	public void setPattern(String pattern) {
		this.pattern = pattern;
	}

}
